package com.atguigu.flink.chapter11.time;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/11/1 08:48
 */
public class Flink_01_Time_Process_1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
       
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        // 2. 在 ddl sql 中
        tEnv.executeSql("create table sensor(" +
                            " id string, " +
                            " ts bigint, " +
                            " vc int, " +
                            " pt as proctime() " + // 定义一个时间字段(时间属性)
                            ")with(" +
                            " 'connector' = 'filesystem', " +
                            " 'path' = 'input/sensor.txt', " +
                            " 'format' = 'csv' " +
                            ")");
    
        tEnv.sqlQuery("select * from sensor").execute().print();
    }
}
